package flink.mysql_connect.cdc;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import flink.mysql_connect.cdc.CdcDataVO;
import io.debezium.data.Envelope;
import io.debezium.time.Timestamp;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.TimeZone;

public class MysqlDataParseSchema implements DebeziumDeserializationSchema<CdcDataVO> {

    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<CdcDataVO> collector) throws Exception {
        CdcDataVO result = new CdcDataVO();
        String topic = sourceRecord.topic();
        String[] split = topic.split("\\.");
        result.setDbName(split[1]);
        result.setTbName(split[2]);

        // 只有after,则表明插入；若只有before，说明删除；若既有before，也有after，则代表更新
        Struct value = (Struct) sourceRecord.value();
        //获取after数据
        Struct after = value.getStruct("after");
        //获取before数据
        Struct before = value.getStruct("before");
        JSONObject beforeJson = new JSONObject();
        if (before != null) {
            beforeJson = getJson(before);
            result.setBeforeMap(beforeJson);
        }
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            //获取列信息
            afterJson = getJson(after);
            result.setAfterMap(afterJson);
        }

        //获取操作类型
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        result.setOperatorType(operation);
        collector.collect(result);

    }

    private JSONObject getJson(Struct struct) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        sdf.setTimeZone(TimeZone.getTimeZone("GMT+0"));
        JSONObject jsonObject = new JSONObject();
        //获取列信息
        Schema schema = struct.schema();
        List<Field> fieldList = schema.fields();
        for (Field field : fieldList) {
            if (Timestamp.SCHEMA_NAME.equals(field.schema().name())){
                jsonObject.put(field.name(),sdf.format(new Date((Long) struct.get(field))));
            }else {
                jsonObject.put(field.name(), struct.get(field));
            }
        }
        return jsonObject;
    }


    @Override
    public TypeInformation<CdcDataVO> getProducedType() {
        return BasicTypeInfo.of(CdcDataVO.class);
    }
}
